import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.util.Collector

/**
 * 需求：
 * 监控温度传感器的温度值，
 * 如果温度值在一秒钟之内(processing time)连续上升，
 * 则报警。
 */
object KeyedProcessFunctionTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream: DataStream[String] = env.socketTextStream("localhost", 7777)

    val tempDS = stream.map(
      data => {
        val dataArr = data.split(",")
        TemperatureData(dataArr(0), dataArr(1).toLong, dataArr(2).toDouble)
      }
    )
    tempDS.keyBy(_.id)
      //TODO TempIncreaseAlertFunction
      .process(new TempIncreaseAlertFunction)


  }
  case class TemperatureData(id: String, time: Long, temp: Double)

  class TempIncreaseAlertFunction extends KeyedProcessFunction[String, TemperatureData, String]{
    //保存上一个传感器温度
    lazy val lastTemp: ValueState[Double] = getRuntimeContext.getState(
      new ValueStateDescriptor[Double]("lastTemp", Types.of[Double])
    )
    //保存注册的定时器的时间戳
    lazy val currentTimer: ValueState[Long] = getRuntimeContext.getState(
      new ValueStateDescriptor[Long]("timer", Types.of[Long])
    )

    override def processElement(value: TemperatureData,
                                ctx: KeyedProcessFunction[String, TemperatureData, String]#Context,
                                out: Collector[String]): Unit = {
      //取出上一次的温度
      val prevTemp: Double = lastTemp.value()
      //将当前温度更新到上一次的温度这个变量中
      lastTemp.update(value.temp)

      val curTimerTimestamp = currentTimer.value()
      if (prevTemp == 0.0 || value.temp < prevTemp){
        //温度下降或者是第一个温度值，删除定时器
        ctx.timerService().deleteProcessingTimeTimer(curTimerTimestamp)
        //清空状态变量
        currentTimer.clear()
      }else if (value.temp > prevTemp && curTimerTimestamp == 0){
        //温度上升且我们并没有设置定时器
        val timerTs = ctx.timerService().currentProcessingTime() + 1000
        ctx.timerService().registerProcessingTimeTimer(timerTs)
        currentTimer.update(timerTs)
      }
    }

    override def onTimer(timestamp: Long,
                         ctx: KeyedProcessFunction[String, TemperatureData, String]#OnTimerContext,
                         out: Collector[String]): Unit = {
      out.collect("传感器 id 为: " + ctx.getCurrentKey + "的传感器温度值已经连续 1s 上升了")
      currentTimer.clear()
    }
  }
}
